package com.atguigu.gamll.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gamll.realtime.app.dwd.BaseApp;
import com.atguigu.gamll.realtime.app.func.DimAsyncFunction;
import com.atguigu.gamll.realtime.beans.OrderDetail;
import com.atguigu.gamll.realtime.beans.OrderInfo;
import com.atguigu.gamll.realtime.beans.OrderWide;
import com.atguigu.gamll.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

// 订单宽表计算
public class OrderWideApp extends BaseApp {
    public static void main(String[] args) throws Exception {
        // TODO 1.基本环境的准备
        // TODO 2.检查点相关的设置
        OrderWideApp orderWideApp = new OrderWideApp();
        orderWideApp.baseEntry();
    }

    @Override
    public void biz(StreamExecutionEnvironment env) {
        // TODO 3.从kafka中读取数据
        String orderInfoTopic = "dwd_order_info";
        String orderDetailTopic = "dwd_order_detail";
        String orderWideGroupId = "orderWideGroupId";

        FlinkKafkaConsumer<String> orderInfoKafka = MyKafkaUtil.getKafkaSource(orderInfoTopic, orderWideGroupId);
        FlinkKafkaConsumer<String> orderDetailKafka = MyKafkaUtil.getKafkaSource(orderDetailTopic, orderWideGroupId);

        DataStreamSource<String> orderInfoStrDS = env.addSource(orderInfoKafka);
        DataStreamSource<String> orderDetailStrDS = env.addSource(orderDetailKafka);

//        orderInfoStrDS.print(">>>");
//        orderDetailStrDS.print("---");

        // TODO 4.对读取的数据进行类型转换  jsonStr --> 实体类对象
        // orderInfo
        SingleOutputStreamOperator<OrderInfo> orderInfoDS = orderInfoStrDS.map(
                new RichMapFunction<String, OrderInfo>() {
                    private SimpleDateFormat sdf;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    }

                    @Override
                    public OrderInfo map(String jsonStr) throws Exception {
                        OrderInfo orderInfo = JSON.parseObject(jsonStr, OrderInfo.class);
                        orderInfo.setCreate_ts(sdf.parse(orderInfo.getCreate_time()).getTime());
                        return orderInfo;
                    }
                }
        );
        // orderDetail
        SingleOutputStreamOperator<OrderDetail> orderDetailDS = orderDetailStrDS.map(
                new RichMapFunction<String, OrderDetail>() {
                    private SimpleDateFormat sdf;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    }
                    @Override
                    public OrderDetail map(String jsonStr) throws Exception {
                        OrderDetail orderDetail = JSON.parseObject(jsonStr, OrderDetail.class);
                        orderDetail.setCreate_ts(sdf.parse(orderDetail.getCreate_time()).getTime());
                        return orderDetail;
                    }
                }
        );

        // TODO 5.指定watermark以及提取事件时间字段
        // 5.1 订单
        SingleOutputStreamOperator<OrderInfo> orderInfoWithWatermarkDS = orderInfoDS.assignTimestampsAndWatermarks(WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<OrderInfo>() {
                    @Override
                    public long extractTimestamp(OrderInfo orderInfo, long l) {
                        return orderInfo.getCreate_ts();
                    }
                })
        );
        // 5.2 订单明细
        SingleOutputStreamOperator<OrderDetail> orderDetailWithWatermarkDS = orderDetailDS.assignTimestampsAndWatermarks(WatermarkStrategy.<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<OrderDetail>() {
                    @Override
                    public long extractTimestamp(OrderDetail orderDetail, long l) {
                        return orderDetail.getCreate_ts();
                    }
                })
        );

        // TODO 6.使用keyBy进行分组 -- 指定两条流的连接字段
        KeyedStream<OrderInfo, Long> orderInfoKeyedDS = orderInfoWithWatermarkDS.keyBy(OrderInfo::getId);
        KeyedStream<OrderDetail, Long> orderDetailKeyedDS = orderDetailWithWatermarkDS.keyBy(OrderDetail::getOrder_id);

        // TODO 7.订单和订单明细进行双流join -- interval join
        SingleOutputStreamOperator<OrderWide> joinedDS = orderInfoKeyedDS
                .intervalJoin(orderDetailKeyedDS)
                .between(Time.seconds(-5), Time.seconds(5))
                .process(
                        new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                            @Override
                            public void processElement(OrderInfo orderInfo, OrderDetail orderDetail, Context ctx, Collector<OrderWide> out) throws Exception {
                                out.collect(new OrderWide(orderInfo, orderDetail));
                            }
                        }
                );
        // joinedDS.print(">>>");

        // TODO 8.关联用户维度
        // unorderedWait：对数据的顺序没有要求
        SingleOutputStreamOperator<OrderWide> orderWideWithUserInfoDS = AsyncDataStream.unorderedWait(
                joinedDS,
                new DimAsyncFunction<OrderWide>("dim_user_info") {
                    @Override
                    public void join(OrderWide orderWide, JSONObject userInfoJsonObj) throws ParseException {
                        orderWide.setUser_gender(userInfoJsonObj.getString("GENDER"));
                        String birthday = userInfoJsonObj.getString("BIRTHDAY");
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                        Long birthdayTime = sdf.parse(birthday).getTime();
                        Long curTime = System.currentTimeMillis();
                        Long ageLone = (curTime - birthdayTime) / 1000L / 60L / 60L / 24L / 365L;
                        orderWide.setUser_age(ageLone.intValue());
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getUser_id().toString();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
//        orderWideWithUserInfoDS.print(">>>");

        // TODO 9.关联地区维度
        SingleOutputStreamOperator<OrderWide> orderWideBaseProvinceDS = AsyncDataStream.unorderedWait(
                orderWideWithUserInfoDS,
                new DimAsyncFunction<OrderWide>("dim_base_province") {
                    @Override
                    public void join(OrderWide orderWide, JSONObject provinceJsonObj) throws Exception {
                        orderWide.setProvince_name(provinceJsonObj.getString("NAME"));
                        orderWide.setProvince_area_code(provinceJsonObj.getString("AREA_CODE"));
                        orderWide.setProvince_iso_code(provinceJsonObj.getString("ISO_CODE"));
                        orderWide.setProvince_3166_2_code(provinceJsonObj.getString("ISO_3166_2"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getProvince_id().toString();
                    }
                },
                60, TimeUnit.SECONDS
        );
//        orderWideBaseProvinceDS.print(">>>");

        // TODO 10.关联商品维度
        SingleOutputStreamOperator<OrderWide> orderWideSpuInfoDS = AsyncDataStream.unorderedWait(
                orderWideBaseProvinceDS,
                new DimAsyncFunction<OrderWide>("dim_sku_info") {
                    @Override
                    public void join(OrderWide orderWide, JSONObject jsonObject) throws Exception {
                        orderWide.setSku_name(jsonObject.getString("SKU_NAME"));
                        orderWide.setCategory3_id(jsonObject.getLong("CATEGORY3_ID"));
                        orderWide.setSpu_id(jsonObject.getLong("SPU_ID"));
                        orderWide.setTm_id(jsonObject.getLong("TM_ID"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getSku_id().toString();
                    }
                },
                60, TimeUnit.SECONDS
        );

        // TODO 11.关联spu
        SingleOutputStreamOperator<OrderWide> orderWideWithSpuDS = AsyncDataStream.unorderedWait(
                orderWideSpuInfoDS, new DimAsyncFunction<OrderWide>("DIM_SPU_INFO") {
                    @Override
                    public void join(OrderWide orderWide, JSONObject jsonObject) throws Exception {
                        orderWide.setSpu_name(jsonObject.getString("SPU_NAME"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getSpu_id());
                    }
                }, 60, TimeUnit.SECONDS);

        // TODO 12.关联品类维度
        SingleOutputStreamOperator<OrderWide> orderWideWithCategory3DS = AsyncDataStream.unorderedWait(
                orderWideWithSpuDS, new DimAsyncFunction<OrderWide>("DIM_BASE_CATEGORY3") {
                    @Override
                    public void join(OrderWide orderWide, JSONObject jsonObject) throws Exception {
                        orderWide.setCategory3_name(jsonObject.getString("NAME"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getCategory3_id());
                    }
                }, 60, TimeUnit.SECONDS);

        // TODO 13.关联品牌维度
        SingleOutputStreamOperator<OrderWide> orderWideWithTmDS = AsyncDataStream.unorderedWait(
                orderWideWithCategory3DS, new DimAsyncFunction<OrderWide>("DIM_BASE_TRADEMARK") {
                    @Override
                    public void join(OrderWide orderWide, JSONObject jsonObject) throws Exception {
                        orderWide.setTm_name(jsonObject.getString("TM_NAME"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getTm_id());
                    }
                }, 60, TimeUnit.SECONDS);

        orderWideWithTmDS.print(">>>");

        // TODO 14.将订单宽表写道kafka中
        orderWideWithTmDS
                .map(JSON::toJSONString)
                .addSink(MyKafkaUtil.getKafkaSink("dwm_order_wide"));
    }
}
